from kafka import KafkaProducer
import time
import random


# 生成数据
def generate_student_info():
    class_id = random.randint(202001, 202005)
    name = random.choice(["Jack", "Luce", "Keven"]) + str(random.randint(1, 100))
    gender = random.choice([0, 1])  # 0代表男生 1代表女生
    birth_date = f"{random.randint(2000, 2004)}-{random.randint(1, 12)}-{random.randint(1, 28)}"
    student_id = "S" + str(random.randint(1000000000, 1900000000))
    semester = str(random.randint(1, 8))
    enrollment_status = random.choice(["L", "A"])
    return f"{class_id}\t{name}\t{gender}\t{birth_date}\t{student_id}\t{semester}\t{enrollment_status}"


def produce_messages(bootstrap_server, topic_name):
    producer = KafkaProducer(bootstrap_servers=bootstrap_server)

    try:
        while True:
            student_info = generate_student_info()
            print(student_info)
            producer.send(topic_name, value=student_info.encode('utf-8'))
            time.sleep(5)  # 五秒产生一组数据
    except KeyboardInterrupt:
        pass

    producer.close()


if __name__ == "__main__":
    bootstrap_servers = '192.168.88.128:9092'
    topic = 'student'
    produce_messages(bootstrap_servers, topic)
